package com.bw.gmall.realtime.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TradeUserSpuOrderBean;
import com.bw.gmall.realtime.utils.DimUtil;
import com.bw.gmall.realtime.utils.DruidDSUtil;
import com.bw.gmall.realtime.utils.JdbcUtil;
import com.bw.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.SQLException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class AsyncDatabaseRequest<T>  extends RichAsyncFunction<T,T> implements  DimJoinInterface<T>{
    private   String  tableName;
    private ThreadPoolExecutor threadPoolExecutor;
    private  DruidDataSource druidDataSource;

    public AsyncDatabaseRequest(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        threadPoolExecutor= ThreadPoolUtil.getThreadPoolExecutor();
        druidDataSource= DruidDSUtil.createDataSource();

    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {

        threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {

                try {
                    DruidPooledConnection connection = druidDataSource.getConnection();

                    JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, getKey(input));
                    //sku_info

                    join(input,dimInfo);
                    connection.close();
                    resultFuture.complete(Collections.singleton(input));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });


    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("超时:"+input);
    }
}
